package com.gitee.cnlongs.cnlongmq.server.operator;

import java.io.File;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.net.URL;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.gitee.cnlongdb.LongDbServer;
import com.gitee.cnlongdb.basic.constant.LongDbConstant;
import com.gitee.cnlongdb.basic.entity.matchingrule.MatchingField;
import com.gitee.cnlongdb.basic.entity.query.LongDbQuery;
import com.gitee.cnlongdb.basic.entity.query.LongDbWhere;
import com.gitee.cnlongdb.basic.utils.RandomUtil;
import com.gitee.cnlongdb.server.config.LongdbConfigHandler;
import com.gitee.cnlongs.cnlongmq.basic.cnlongds.annotation.LongdsTask;
import com.gitee.cnlongs.cnlongmq.basic.cnlongds.entity.CnLongdsMethodEntity;
import com.gitee.cnlongs.cnlongmq.basic.cnlongds.utils.AnnotationScannerUtils;
import com.gitee.cnlongs.cnlongmq.basic.config.LongmqConfigHandler;
import com.gitee.cnlongs.cnlongmq.basic.constant.CnLongMqConstant;
import com.gitee.cnlongs.cnlongmq.basic.constant.LongMqDataMap;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMessageQueue;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMqConfiger;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMqMessage;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.response.CnLongMqResponse;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqDataSync;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqMode;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqRole;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqStatus;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqType;
import com.gitee.cnlongs.cnlongmq.basic.enums.exception.CnLongMqExceptionEnums;
import com.gitee.cnlongs.cnlongmq.basic.exception.CnLongMqBaseException;
import com.gitee.cnlongs.cnlongmq.server.engine.LongMqEngine;
import com.gitee.cnlongs.cnlongmq.server.handler.LongMqRequest;
import com.gitee.cnlongs.cnlongmq.server.threads.ElectionCaptainSequeueHandler;
import com.gitee.cnlongs.cnlongmq.server.threads.LongMqDataSyncHandler;

/**
 * 系统名称：CnLongMqOperator.java
 * 模块名称：
 * 模块描述：中国龙[消息队列] - 执行器
 * 功能列表：
 * 模块作者：ZOUYONG
 * 开发时间：2019年3月31日 下午3:39:15
 * 模块路径：com.gitee.cnlongs.cnlongmq.server.operator.CnLongMqExecuter
 */
public class CnLongMqExecuter {

	private final Logger logger = (Logger) LoggerFactory.getLogger(CnLongMqExecuter.class);

	// mq消息配置对象
	private CnLongMqConfiger longMqConfig;
	
	// mq消息数据库操作对象
	private LongDbServer longDbServer;
	
	public CnLongMqExecuter(String cnlongMqConfigPath) {
		// 加载mq配置文件
		this.longMqConfig = loadLongMqConfig(cnlongMqConfigPath);
		// 开启消息队列数据库监听
		this.longDbServer = new LongDbServer(this.longMqConfig.getLongMqPath(), this.longMqConfig.getLongMqName(), 
				this.longMqConfig.getLongMqIp(), this.longMqConfig.getLongMqPort(), this.longMqConfig.getDbConfigPath());
		// 设置队列表数据分区规则
		setSequeueMatchingData();
		// 执行队列信息交换
		LongMqDataSyncHandler.handlerNeedSequeueShare(this.longMqConfig, null);
		// 开启各种监听操作
		initAssistActuator();
		// 开启队长选举操作
		ElectionCaptainSequeueHandler.startElectionCaptainOperator(longMqConfig);
		// 开启扫码定时任务注解功能
		// startScanTimingTask();
	}

	/**
	 * 功能描述：开启扫码定时任务注解功能
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年8月1日 上午10:35:52
	 × 返回数据：void
	 */
	private void startScanTimingTask() {
		Map<Class<?>, List<Method>> clsMethMap = new HashMap<Class<?>, List<Method>>();
		String pkgNames = "com.gitee.cnlongs.cnlongmq.server.test.task;com.gitee.cnlongs.cnlongmq.basic.cnlongds.task";
		String[] pkgNameAry = pkgNames.split(";");
		for (String pkgName : pkgNameAry) {
			String pkgPath = AnnotationScannerUtils.getPkgPath(pkgName);
			clsMethMap.putAll(AnnotationScannerUtils.scanClassesByAnnotations(pkgName, pkgPath, true, LongdsTask.class));
		}

		List<CnLongdsMethodEntity> methodList = new ArrayList<CnLongdsMethodEntity>();
		for(Iterator<Class<?>> iter = clsMethMap.keySet().iterator(); iter.hasNext();) {
			Class<?> curClass = iter.next();
			List<Method> methods = clsMethMap.get(curClass);
			for (Method method : methods) {
				CnLongdsMethodEntity methodObj = new CnLongdsMethodEntity();
				Annotation[] tations = method.getAnnotations();
				for (Annotation annotation : tations) {
					try {
						LongdsTask task = (LongdsTask) annotation;
						methodObj.setFormat(task.format());
						break;
					} catch(Exception e) {
						logger.error("转换定时任务方法注解内容异常：{}", e.getMessage());
					}
				}
				if(StringUtils.isNotBlank(methodObj.getFormat())) {
					methodObj.setMethod(method);
					methodObj.setClasss(curClass);
					methodObj.setParameterTypes(method.getParameterTypes());
					methodList.add(methodObj);
					// 利用java反射调用方法
					// method.invoke(curClass.newInstance(), method.getParameterTypes());
				}
			}
		}
	}

	/**
	 * 功能描述：设置队列表数据分区规则
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月15日 下午4:13:37
	 × 返回数据：void
	 */
	private void setSequeueMatchingData() {
		if(StringUtils.isNotBlank(this.longMqConfig.getDataTimeMatching())) {
			this.longDbServer.getConnect().setDateTimeMatching(
					this.longDbServer.getConnect().getDataBaseName(), this.longMqConfig.getDataTimeMatching(), true);
		}
	}

	/**
	 * 功能描述：初始化各种协助器
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年5月5日 下午3:00:59
	 × 返回数据：void
	 */
	private void initAssistActuator() {
		// 初始化执行引擎
		LongMqEngine.startEngineMonitor(this.longDbServer.getConnect());
	}

	/**
	 * 功能描述：加载mq配置文件
	 * 模块作者：ZOUYONG
	 * 开发时间：2019年3月31日 下午4:19:46
	 * 返回数据：void
	 */
	private CnLongMqConfiger loadLongMqConfig(String cnlongMqConfigPath) {
		
		// 校验MQ配置文件信息
		File[] files = checkMqConfigInfo(cnlongMqConfigPath);
		
		File cfgFile = null;
		List<File> seqFileList = new ArrayList<File>();
		for (File fi : files) {
			// 获取总配置文件
			if(fi.isFile() && fi.getName().endsWith(CnLongMqConstant.longmqCfgSuffix)) {
				cfgFile = fi;
			}
			// 获取队列配置文件列表
			else if(fi.isFile() && fi.getName().endsWith(CnLongMqConstant.longmqSeqSuffix)) {
				seqFileList.add(fi);
			}
		}
		
		if(null == cfgFile) {
			throw new RuntimeException("未加载到" + CnLongMqConstant.longmqDefaultName + "配置文件！");
		}
		
		CnLongMqConfiger mqConfig = LongmqConfigHandler.getLongMqConfigInfo(cfgFile);
		mqConfig.check(); // 校验属性是否正确
		
		// 获取队列列表
		for (File seqFile : seqFileList) {
			List<CnLongMessageQueue> sequeueList = LongmqConfigHandler.getLongMqSeq(mqConfig, seqFile);
			mqConfig.getSequeueList().addAll(sequeueList);
			for (CnLongMessageQueue cnLongMqSeqEntity : sequeueList) {
				try {
					CnLongMessageQueue mqSeq = (CnLongMessageQueue) cnLongMqSeqEntity.clone();
					mqConfig.getCustomerSequeueList().add(mqSeq);
				} catch (Exception e) {
					throw new RuntimeException(e);
				}
			}
		}
		
		return mqConfig;
	}

	/**
	 * 功能描述：校验配置文件是否合法
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月18日 下午12:15:01
	 × 返回数据：File[]
	 */
	private File[] checkMqConfigInfo(String cnlongMqConfigPath) {
		logger.debug(CnLongMqConstant.longmqDefaultName + "开始校验配置文件：{}", cnlongMqConfigPath);
		if(StringUtils.isBlank(cnlongMqConfigPath)) {
			throw new RuntimeException(CnLongMqConstant.longmqDefaultName + "配置目录不能为空！");
		}
		URL url = this.getClass().getClassLoader().getResource(cnlongMqConfigPath);
		if(null == url) {
			throw new RuntimeException("未加载到" + CnLongMqConstant.longmqDefaultName + "的配置信息！");
		}
		File file = new File(url.getPath());
		if(!file.exists()) {
			throw new RuntimeException("未加载到" + CnLongMqConstant.longmqDefaultName + "配置目录中没有文件！");
		}
		File[] files = file.listFiles();
		if(null == files || files.length == 0) {
			throw new RuntimeException("未加载到" + CnLongMqConstant.longmqDefaultName + "配置目录为空！");
		}
		return files;
	}

	/**
	 * 功能描述：处理发送mq消息的请求
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月12日 下午5:00:23
	 × 返回数据：String
	 */
	public CnLongMqResponse handlerSendLongMqMessage(String seqKey, Object message) throws Exception {
		
		// 将消息对象转换成JSON格式
		String longMqText = null != message ? JSON.toJSONString(message) : "";
		if(StringUtils.isBlank(longMqText)) {
			throw new RuntimeException("要发送的消息内容不能为空！");
		}

		// 定义发送消息的对象
		CnLongMqMessage mqMessage = new CnLongMqMessage();
		mqMessage.setLongMqText(longMqText);
		mqMessage.setLongMqKey(seqKey);

		return againSendLongMqMessage(mqMessage, true);
	}

	/**
	 * 功能描述：消息发送 或 重发
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月18日 下午3:24:47
	 × 返回数据：void
	 */
	public CnLongMqResponse againSendLongMqMessage(CnLongMqMessage mqMessage, boolean isRecordRes) {
		
		if(null == mqMessage) { throw new RuntimeException("要发送的消息对象不能为空！"); }
		
		List<CnLongMessageQueue> execuSequeueList = new ArrayList<CnLongMessageQueue>();
		CnLongMessageQueue sequeue = null;
		try {
			sequeue = checkSendLongMqMessage(mqMessage, execuSequeueList);
		} catch(Exception e) {
			if(e instanceof CnLongMqBaseException) {
				CnLongMqBaseException baseExp = (CnLongMqBaseException) e;
				if(baseExp.getExpNums() == CnLongMqExceptionEnums.not_cansend_sequence_list && !mqMessage.isAgainSend()) {
					System.out.println("消息发送失败，没有可用的队列数据[{}]！" + mqMessage.getLongMqKey());
					return processSendFailMqMessage(null, baseExp.getSequence(), mqMessage);
				}
			}
			throw e;
		}
		
		String longMqText = mqMessage.getLongMqText();
		if(StringUtils.isBlank(longMqText)) {
			throw new RuntimeException("要发送的消息内容不能为空！");
		}
		
		// 发送方-服务器信息
		mqMessage.setLongMqStatus(CnLongMqStatus.processing.name());
		mqMessage.setLongMqKey(sequeue.getKey());
		mqMessage.setSendServerName(longMqConfig.getLongMqName());
		mqMessage.setSendServerIp(longMqConfig.getLongMqIp());
		mqMessage.setSendServerPort(longMqConfig.getLongMqPort());
		mqMessage.setLongMqRole(CnLongMqRole.provider.name());
		mqMessage.setSendDate(new Date());
		mqMessage.setLongMqTimeOut(sequeue.getTimeout());

		CnLongMqResponse response = null;
		// 如果是点对点
		if(CnLongMqType.point.name().equals(sequeue.getType())) {
			response = sendPointMqMessage(mqMessage, execuSequeueList);
			if(isRecordRes && (null == response || !response.getReturnStatus())) {
				// 发送失败，保存消息进行补偿处理
				response = processSendFailMqMessage(response, sequeue, mqMessage);
			}
		}
		// 如果是广播
		else if(CnLongMqType.broad.name().equals(sequeue.getType())) {
			response = new CnLongMqResponse();
			for (CnLongMessageQueue mqSeq : execuSequeueList) {
				if(!mqMessage.isAgainSend()) {
					mqMessage.setLongMqId(null); // 清空消息ID，目的是多次发送失败，可以重复发送
				}
				mqMessage.setAgainSend(false);
				// 消费方信息
				mqMessage.setCustomerServerName(mqSeq.getLongMqName());
				mqMessage.setCustomerServerIp(mqSeq.getLongMqIp());
				mqMessage.setCustomerPort(mqSeq.getLongMqPort());
				mqMessage.setClassPath(mqSeq.getClassPath());
				// 开始发送消息
				CnLongMqResponse oneResp = startSendLongMqMessage(mqMessage);
				if(isRecordRes && (null == oneResp || !oneResp.getReturnStatus())) {
					// 发送失败，保存消息进行补偿处理
					processSendFailMqMessage(oneResp, sequeue, mqMessage);
					response.setSendFailNumber(response.getSendFailNumber()+1);
					response.setReturnStatus(false);
				} else {
					response.setSendSuccNumber(response.getSendSuccNumber()+1);
				}
			}
		}
		
		return response;
	}

	/**
	 * 功能描述：校验发送消息数据的正确性
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月18日 下午3:34:04
	 × 返回数据：CnLongMessageQueue
	 */
	private CnLongMessageQueue checkSendLongMqMessage(CnLongMqMessage mqMessage, List<CnLongMessageQueue> execuSequeueList) {
		if(null == mqMessage || StringUtils.isBlank(mqMessage.getLongMqKey())) {
			throw new RuntimeException("要发送消息的MQ队列名称不能为空！");
		}
		
		CnLongMessageQueue sequeue = getSequeueByKey(mqMessage.getLongMqKey(), this.longMqConfig);
		if(null == sequeue) {
			throw new RuntimeException("没有找到队列的基础数据信息！");
		}
		
		// 获取消费者的消息队列列表
		List<CnLongMessageQueue> customerSequeueList = this.longMqConfig.getCustomerSequeueList(CnLongMqRole.customer.name());
		for (CnLongMessageQueue mqSeq : customerSequeueList) {
			// 判断队列是否已经发送成功
			String mapKey = mqSeq.getLongMqIp() + CnLongMqConstant.longmqSeparateF + mqSeq.getLongMqPort();
			if(null == mqMessage.getSendCustomerMap().get(mapKey) || !mqMessage.getSendCustomerMap().get(mapKey))
			{
				if(mqSeq.judgeSequeueIsAvailable(sequeue)) {
					execuSequeueList.add(mqSeq);
				}
			}
		}
		
		// 如果是重发，则要剔除【已发送成功】的队列数据
		if(mqMessage.isAgainSend() && CollectionUtils.isNotEmpty(execuSequeueList)) {
			LongDbQuery query = new LongDbQuery();
			query.setTableName(mqMessage.getLongMqKey());
			query.setClassz(CnLongMqMessage.class);
			query.setPageSize(999999);
			query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqStatus.name()).is(CnLongMqStatus.success.name()).tableMatching(true));
			query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqMsgId.name()).is(mqMessage.getLongMqMsgId()));
			try {
				List<CnLongMqMessage> list = longDbServer.find(query);
				if(CollectionUtils.isNotEmpty(list)) {
					for (int i=0; i<execuSequeueList.size(); i++) {
						CnLongMessageQueue queue = execuSequeueList.get(i);
						for (CnLongMqMessage mqSeq : list) {
							if(queue.compareByMqMessage(mqSeq)) {
								execuSequeueList.remove(i);
								i--;
							}
						}
					}
				}
			} catch (Exception e) {
				logger.error("查询已发送成功的队列信息异常：{}", e.getMessage());
			}
		}
		
		if(CollectionUtils.isEmpty(execuSequeueList)) {
			logger.debug("没有找到消费者队列信息！");
			throw new CnLongMqBaseException(CnLongMqExceptionEnums.not_cansend_sequence_list, sequeue);
		}
		
		return sequeue;
	}

	/**
	 * 功能描述：对处理中的消息进行补偿处理
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年7月26日 下午6:31:54
	 × 返回数据：void
	 */
	public void handProcessingMessage(CnLongMqMessage mqMessage) {
		if(null == mqMessage || StringUtils.isNotBlank(
				LongMqDataMap.sendProcessingMqMessage.get(mqMessage.getLongMqId()))) {
			return; // 消息正在发送，则停止重试操作
		}
		
		// 如果发送方就是当前应用，则重新发送
		if(mqMessage.getSendServerName().equals(this.longMqConfig.getLongMqName()) && 
				mqMessage.getSendServerIp().equals(this.longMqConfig.getLongMqIp()) && 
				mqMessage.getSendServerPort().equals(this.longMqConfig.getLongMqPort())
		) {
			this.againSendLongMqMessage(mqMessage, false);
		}
		// 否：调用消息响应方法，并且发起ACK确认
		else {
			CnLongMqResponse message = new CnLongMqResponse();
			message.setMessageContent(mqMessage.getLongMqText());
			message.setLongMqKey(mqMessage.getLongMqKey());
			mqMessage.setReqType(CnLongMqDataSync.noticeAckResult.name());
			try {
				LongMqRequest.launchSendMessage(mqMessage);
			} catch (Exception e) {
				throw new RuntimeException("调用业务系统响应消息的方法，执行ACK确认异常：" + e.getMessage());
			}
		}
	}

	/**
	 * 功能描述：对发送失败的消息进行补偿处理
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月18日 下午2:22:13
	 × 返回数据：void
	 */
	private CnLongMqResponse processSendFailMqMessage(
			CnLongMqResponse response, CnLongMessageQueue sequeue, CnLongMqMessage mqMessage) {
		
		if(null == response) {
			response = new CnLongMqResponse();
		}
		
		// 对发送失败消息的基本信息继续补充
		mqMessage = supplementMqMessageBaseInfo(sequeue, mqMessage);
		
		response.setReturnStatus(false);
		String errMsg = response.getReturnContent();
		
		// 如果是点对点
		if(CnLongMqType.point.name().equals(sequeue.getType())) {
			if(StringUtils.isBlank(errMsg)) {
				errMsg = "同步发送MQ点对点消息失败，没有Customer正常消费！";
			}
		}
		// 如果是广播
		else if(CnLongMqType.broad.name().equals(sequeue.getType())) {
			if(StringUtils.isBlank(errMsg)) {
				errMsg = "同步发送MQ广播消息失败，请确认每个Customer运行是否正常！";
			}
		}
		response.setReturnContent(errMsg);
		
		// 如果是同步发送，暂时不需要补偿处理，直接将消【息废弃调】
		if(CnLongMqMode.sync.name().equals(sequeue.getMode())) {
			LongMqRequest.processTransferMqMessage(mqMessage, CnLongMqStatus.discard.name());
		}
		// 如果是异步发送，则转移消息至【处理失败】，开启自动定时重发，直到消息过期时间，或者人工删除消息
		else if(CnLongMqMode.async.name().equals(sequeue.getMode())) {
			LongMqRequest.processTransferMqMessage(mqMessage, CnLongMqStatus.failure.name());
		}
		
		return response;
	}

	/**
	 * 功能描述：对失败消息的基本信息继续补充
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年7月25日 下午7:55:15
	 × 返回数据：CnLongMqMessage
	 */
	private CnLongMqMessage supplementMqMessageBaseInfo(CnLongMessageQueue sequeue, CnLongMqMessage mqMessage) {
		
		// 发送日期
		if(null == mqMessage.getSendDate()) {
			mqMessage.setSendDate(new Date());
		}
		
		// 消息过期时间
		if(mqMessage.getLongMqTimeOut() == 0) {
			mqMessage.setLongMqTimeOut(sequeue.getTimeout());
		}
		
		// 消息ID
		setLongMqMessageId(mqMessage, false);
		
		return mqMessage;
	}
	
	/**
	 * 功能描述：设置消息ID和消息内容的ID
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年7月26日 下午6:30:49
	 × 返回数据：CnLongMqMessage
	 */
	private CnLongMqMessage setLongMqMessageId(CnLongMqMessage mqMessage, boolean force) {
		
		// 消息ID
		if(force || StringUtils.isBlank(mqMessage.getLongMqId())) {
			mqMessage.setLongMqId(RandomUtil.getCharAndNumr(32));
		}
		
		// 消息内容ID（可能会对应多个消息ID）
		if(StringUtils.isBlank(mqMessage.getLongMqMsgId())) {
			mqMessage.setLongMqMsgId(RandomUtil.getCharAndNumr(32));
		}
		
		return mqMessage;
	}

	/**
	 * 功能描述：消息点对点发送
	 * 模块作者：ZOUYONG
	 * 开发时间：2019年4月16日 下午8:29:53
	 * 返回数据：void
	 */
	private CnLongMqResponse sendPointMqMessage(CnLongMqMessage mqMessage, List<CnLongMessageQueue> execuSequeueList) {
		if(CollectionUtils.isEmpty(execuSequeueList)) {
			return null;
		}
		// 获取指定范围内的随机数
		Random rand = new Random();
		int numSize = rand.nextInt(execuSequeueList.size());
		
		CnLongMessageQueue mqSeq = execuSequeueList.get(numSize);
		// 消费方信息
		mqMessage.setCustomerServerName(mqSeq.getLongMqName());
		mqMessage.setCustomerServerIp(mqSeq.getLongMqIp());
		mqMessage.setCustomerPort(mqSeq.getLongMqPort());
		mqMessage.setClassPath(mqSeq.getClassPath());
		mqMessage.setCreateDate(new Date());
		// 开始发送消息
		CnLongMqResponse response = startSendLongMqMessage(mqMessage);
		// 如果发送失败，则继续循环发送
		if(null == response || !response.getReturnStatus()) {
			execuSequeueList.remove(numSize);
			sendPointMqMessage(mqMessage, execuSequeueList);
		}
		return response;
	}

	/**
	 * 功能描述：开始发送消息
	 * 模块作者：ZOUYONG
	 * 开发时间：2019年4月13日 上午10:55:31
	 * 返回数据：void
	 */
	private CnLongMqResponse startSendLongMqMessage(CnLongMqMessage mqMessage) {
		try {
			// 标记该Mq消息正在处理
			LongMqDataMap.sendProcessingMqMessage.put(mqMessage.getLongMqId(), mqMessage.getLongMqId());
			mqMessage.setReqType(CnLongMqDataSync.sendLongMqMessage.name());
			try {
				if(StringUtils.isBlank(mqMessage.getLongMqId())) {
					setLongMqMessageId(mqMessage, false);
					// 保存消息数据
					saveMqMessage(mqMessage);
				} else {
					// 更新消息数据
					updtMqMessage(mqMessage);
				}
			} catch (Exception e) {
				logger.error("持久化消息异常：" + e.getMessage());
				throw new RuntimeException(e);
			}
			try {
				// 发起消息推送
				return LongMqRequest.launchSendMessage(mqMessage);
			}
			catch (Exception e) {
				throw new RuntimeException("消息推送异常：" + e.getMessage());
			}
		}
		finally {
			// 清除该Mq消息，处理完成
			LongMqDataMap.sendProcessingMqMessage.remove(mqMessage.getLongMqId());
		}
	}

	/**
	 * 功能描述：保存MQ消息到磁盘
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月17日 上午11:16:42
	 × 返回数据：void
	 */
	public void saveMqMessage(CnLongMqMessage mqMessage) {
		if(null == mqMessage) {
			throw new RuntimeException("要保存的消息数据不能为空！");
		}
		try {
			// 添加MQ表数据的主外键分区规则
			addMqMessageMatching(mqMessage);
			// 将消息保存到磁盘
			longDbServer.insert(mqMessage, mqMessage.getLongMqKey());
		} catch (Exception e) {
			throw new RuntimeException("保存消息失败！" + e.getMessage());
		}
	}

	/**
	 * 功能描述：添加MQ表数据的主外键分区规则
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月17日 上午10:34:45
	 × 返回数据：void
	 */
	public void addMqMessageMatching(CnLongMqMessage mqMessage) {

		List<Object> matchList = LongdbConfigHandler
				.getMatchingRuleListByTableName(this.longDbServer.getConnect(), mqMessage.getLongMqKey());
		if(CollectionUtils.isEmpty(matchList)) {
			matchList = new ArrayList<Object>();
		}
		for (Object object : matchList) {
			if(object instanceof MatchingField) {
				MatchingField field = (MatchingField) object;
				List<String> fds = field.getFieldMap().get(mqMessage.getLongMqKey());
				if(CollectionUtils.isNotEmpty(fds)) {
					return;
				}
			}
		}
		
		// 设置按【消息同步状态】进行分区
		List<String> fds = new ArrayList<String>();
		fds.add(CnLongMqDataSync.longMqStatus.name());

		MatchingField field = new MatchingField();
		field.setRuleType(LongDbConstant.fieldMatching);
		field.getFieldMap().put(mqMessage.getLongMqKey(), fds);
		this.longDbServer.getConnect().getMatchingList().add(field);
	}

	/**
	 * 功能描述：获取指定Key值的队列信息
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月16日 下午2:58:35
	 × 返回数据：CnLongMqSeqEntity
	 */
	public static CnLongMessageQueue getSequeueByKey(String seqKey, CnLongMqConfiger mqConfig) {
		// 获取配置的队列列表
		List<CnLongMessageQueue> sequeueList = mqConfig.getSequeueList();
		if(CollectionUtils.isEmpty(sequeueList)) {
			throw new RuntimeException("没有找到配置的队列列表信息！");
		}
		
		// 获取队列基础信息
		CnLongMessageQueue sequeue = null;
		for (CnLongMessageQueue mqSeq : sequeueList) {
			if(seqKey.equals(mqSeq.getKey())) {
				sequeue = mqSeq;
				break;
			}
		}
		return sequeue;
	}
	
	/**
	 * 功能描述：更新消息数据
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年5月8日 上午11:26:48
	 × 返回数据：void
	 */
	private void updtMqMessage(CnLongMqMessage mqMessage) {
		try {
			LongDbQuery query = new LongDbQuery();
			query.setTableName(mqMessage.getLongMqKey());
			query.setClassz(CnLongMqMessage.class);
			query.setPageSize(1);
			query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqId.name()).is(mqMessage.getLongMqId()));
			boolean updStatus = longDbServer.update(query, mqMessage);
			logger.debug("更新状态：" + updStatus);
		} catch (Exception e) {
			throw new RuntimeException("更新消息失败！" + e.getMessage());
		}
	}

	public CnLongMqConfiger getLongMqConfig() {
		return longMqConfig;
	}

	public LongDbServer getLongDbServer() {
		return longDbServer;
	}

}
